use std::future::Future;
use std::marker::PhantomData;
use std::mem::ManuallyDrop;
use std::ops::Deref;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::sync::mpsc::{Receiver, sync_channel, SyncSender};
use std::task::{Context, RawWaker, RawWakerVTable, Wake, Waker};

use crossbeam::channel::internal::SelectHandle;
use log::set_logger_racy;

use crate::{safe_pin};
use crate::utils::ThreadSafePtr;

//
// /// Simplified waking interface based on Arcs.
// pub(crate) trait Wake: Send + Sync + Sized + 'static {
//     // Wake by value.
//     fn wake(self: Arc<Self>);
//
//     /// Wake by reference.
//     fn wake_by_ref(self: &Arc<Self>) {
//         self.clone().wake();
//     }
// }
//
// /// A `Waker` that is only valid for a given lifetime.
// #[derive(Debug)]
// pub(crate) struct WakerRef<'a> {
//     waker: ManuallyDrop<Waker>,
//     _p: PhantomData<&'a ()>,
// }
//
// impl Deref for WakerRef<'_> {
//     type Target = Waker;
//
//     fn deref(&self) -> &Waker {
//         &self.waker
//     }
// }
//
// /// Creates a reference to a `Waker` from a reference to `Arc<impl Wake>`.
// pub(crate) fn waker_ref<W: Wake>(wake: &Arc<W>) -> WakerRef<'_> {
//     let ptr = Arc::as_ptr(wake) as *const ();
//
//     let waker = unsafe { Waker::from_raw(RawWaker::new(ptr, waker_vtable::<W>())) };
//
//     WakerRef {
//         waker: ManuallyDrop::new(waker),
//         _p: PhantomData,
//     }
// }
//
// fn waker_vtable<W: Wake>() -> &'static RawWakerVTable {
//     &RawWakerVTable::new(
//         clone_arc_raw::<W>,
//         wake_arc_raw::<W>,
//         wake_by_ref_arc_raw::<W>,
//         drop_arc_raw::<W>,
//     )
// }
//
// unsafe fn inc_ref_count<T: Wake>(data: *const ()) {
//     // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop
//     let arc = ManuallyDrop::new(Arc::<T>::from_raw(data as *const T));
//
//     // Now increase refcount, but don't drop new refcount either
//     let _arc_clone: ManuallyDrop<_> = arc.clone();
// }
//
// unsafe fn clone_arc_raw<T: Wake>(data: *const ()) -> RawWaker {
//     inc_ref_count::<T>(data);
//     RawWaker::new(data, waker_vtable::<T>())
// }
//
// unsafe fn wake_arc_raw<T: Wake>(data: *const ()) {
//     let arc: Arc<T> = Arc::from_raw(data as *const T);
//     Wake::wake(arc);
// }
//
// // used by `waker_ref`
// unsafe fn wake_by_ref_arc_raw<T: Wake>(data: *const ()) {
//     // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop
//     let arc = ManuallyDrop::new(Arc::<T>::from_raw(data as *const T));
//     Wake::wake_by_ref(&arc);
// }
//
// unsafe fn drop_arc_raw<T: Wake>(data: *const ()) {
//     drop(Arc::<T>::from_raw(data as *const T))
// }

type BoxFuture<T> = Pin<Box<dyn Future<Output=T> + Send + 'static>>;

/// 执行器
pub struct Executor {
    ready_queue: Receiver<Arc<Task>>,
    task_sender: Option<SyncSender<Arc<Task>>>,
}

struct Task {
    future_item: Mutex<Option<BoxFuture<()>>>,
    task_sender: SyncSender<Arc<Task>>,
}

impl Wake for Task {
    // 关键步骤，当完成时，重新推入队列
    fn wake(self: Arc<Self>) {
        self.task_sender
            .send(self.clone())
            .expect("full of future_task!!!");
    }
}

impl Executor {
    /// 添加一个future_task
    pub fn spawn(&self, future_task: impl Future<Output=()> + 'static + Send) -> &Executor {
        let task = Arc::new(Task {
            future_item: Mutex::new(Some(Box::pin(future_task))),
            task_sender: self.task_sender.clone().unwrap(),
        });
        match &self.task_sender {
            Some(sender) => sender.send(task).expect("full of future_task!!!"),
            _ => panic!("required sender missed!!!"),
        };
        self
    }
    /// 开启所有future_tasks
    pub fn run(&mut self) {
        self.task_sender.take();
        while let Ok(task) = self.ready_queue.recv() {
            let waker = Waker::from(task.clone());
            // 获取一个future，若它还没有完成(仍然是Some，不是None)，则对它进行一次poll并尝试完成它
            let mut future_slot = task.future_item.lock().unwrap();
            if let Some(mut future) = future_slot.take() {
                // 基于任务自身创建一个 `LocalWaker`
                // `BoxFuture<T>`是`Pin<Box<dyn Future<Output = T> + Send + 'static>>`的类型别名
                // 通过调用`as_mut`方法，可以将上面的类型转换成`Pin<&mut dyn Future + Send + 'static>`
                if future.as_mut().poll(&mut Context::from_waker(&waker)).is_pending() {
                    // Future还没执行完，因此将它放回任务中，等待下次被poll
                    *future_slot = Some(future);
                }
            }
        }
    }
    pub fn new(size: usize) -> Self {
        let (task_sender, ready_queue) = sync_channel(size);
        Self {
            ready_queue,
            task_sender: Some(task_sender),
        }
    }
}

/// 对于async函数的深入理解
/// 一般来说async和await必须成对出现，否者async函数是不会执行的
/// 最外层的async函数必须有触发器调用（例如tokio::runtime运行时）
/// 每一个async函数都会被编译器解包成为一个struct
/// # example
/// ```
/// struct AsyncFuture {
///     fut_one: FutOne,
///     fut_two: FutTwo,
///
///     state: State,
///
/// }
///
/// // `async` 语句块可能处于的状态
/// enum State {
///
///     AwaitingFutOne,
///     AwaitingFutTwo,
///     Done,
/// }
///
///
/// impl Future for AsyncFuture {
///     type Output = ();
///
///     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
///         loop {
///             match self.state {
///                 State::AwaitingFutOne => match self.fut_one.poll(..) {
///                     Poll::Ready(()) => self.state = State::AwaitingFutTwo,
///                     Poll::Pending => return Poll::Pending,
///                 }
///                 State::AwaitingFutTwo => match self.fut_two.poll(..) {
///                     Poll::Ready(()) => self.state = State::Done,
///                     Poll::Pending => return Poll::Pending,
///                 }
///                 State::Done => return Poll::Ready(()),
///             }
///         }
///     }
/// }
/// ```


/// 自引用在future中的使用情况
/// ```
///
/// async {
///     let mut x = [0; 128];
///     let read_into_buf_fut = read_into_buf(&mut x);
///     read_into_buf_fut.await;
///     println!("{:?}", x);
/// }
///
///
/// struct ReadIntoBuf<'a> {
///     buf: &'a mut [u8], // 指向下面的`x`字段
/// }
///
/// struct AsyncFuture {
///     x: [u8; 128],
///     read_into_buf_fut: ReadIntoBuf<'what_lifetime?>,
/// }
///
/// ```
///
///
///
///
///
///
async fn test_move() {
    let mut ok_str = "1234567890".to_string();
    print_str(&mut ok_str).await;
}

async fn print_str(strs: &mut String) {
    println!("{}", strs);
}

#[test]
fn test_executor() {
    let mut executor = Executor::new(10);
    // safe_pin! {
    // let fut = test_move();
    // }
    // let funcp = ThreadSafePtr::capture(&test_move);
    let funcp =test_move;
    executor.spawn(async move{
        for _ in 0..5 {
            funcp().await;
        }
    });

    executor.run();
}